package com.spark.sql

import org.ansj.splitWord.analysis.BaseAnalysis
import org.apache.spark.graphx.Graph
import org.apache.spark.mllib.feature.{HashingTF, IDF}
import org.apache.spark.sql.SparkSession
import org.jsoup.Jsoup
import org.apache.spark.mllib.linalg.{SparseVector => SV}

/**
  * Created by TRS on 2017/6/12.
  */
object newsSQL {
  def main(args: Array[String]): Unit = {

    val sim=0.5

    val sparkSession = SparkSession.builder
      .config("spark.sql.warehouse.dir", "D:\\WorkSpace\\spark\\spark-learning\\spark-warehouse")
      .config("spark.driver.memory","2G")
      .master("local")
      .appName("spark session example")
      .getOrCreate()


    val file="E:\\zjol\\10000\\1.json";


    val df = sparkSession.read.json(file)

    df.printSchema()
    df.createOrReplaceTempView("news")

    val sql="select author,body,is_topic,keywords,newsid,pub_time,source,sub_title,title,top_title,topicurl from news"

    val rdd=sparkSession.sql(sql).rdd.map(row=>
      (
        row.getString(4).substring(1).toLong,
        row.getString(8),
       getTextFromTHML(row.getString(6))
      )
    ).filter(x=>(!x._2.equals("") && !x._3.equals("")))

    rdd.foreach(x=>println(x._1+"===="+x._2+"===="+x._3))





  }


  def getTextFromTHML(htmlStr: String): String = {
    val doc = Jsoup.parse(htmlStr)
    var text1 = doc.text()
    // remove extra white space
    val builder = new StringBuilder(text1)
    var index = 0
    while ( {
      builder.length > index
    }) {
      val tmp = builder.charAt(index)
      if (Character.isSpaceChar(tmp) || Character.isWhitespace(tmp)) builder.setCharAt(index, ' ')
      index += 1
    }
    text1 = builder.toString.replaceAll(" +", " ").trim
    text1
  }



  //相似度比对 最短编辑距离
  def ld(s: String, t: String): Int = {
    var sLen: Int = s.length
    var tLen: Int = t.length
    var cost: Int = 0
    var d = Array.ofDim[Int](sLen + 1, tLen + 1)
    var ch1: Char = 0
    var ch2: Char = 0
    if (sLen == 0)
      tLen
    if (tLen == 0)
      sLen
    for (i <- 0 to sLen) {
      d(i)(0) = i
    }
    for (i <- 0 to tLen) {
      d(0)(i) = i
    }
    for (i <- 1 to sLen) {
      ch1 = s.charAt(i - 1)
      for (j <- 1 to tLen) {
        ch2 = t.charAt(j - 1)
        if (ch1 == ch2) {
          cost = 0
        } else {
          cost = 1
        }
        d(i)(j) = Math.min(Math.min(d(i - 1)(j) + 1, d(i)(j - 1) + 1), d(i - 1)(j - 1) + cost)
      }
    }
    return d(sLen)(tLen)
  }

  /**
    * Levenshtein
    *
    * @param src
    * @param tar
    * @return
    */
  def similarity(src: String, tar: String): Double = {
    val a: Int = ld(src, tar)
    1 - a / (Math.max(src.length, tar.length) * 1.0)
  }

  //选出一组字符串 中相似度最高的
  def mostSimilartyTitle(strs: Array[String]): String = {
    var map: Map[String, Double] = Map()
    for (i <- 0 until strs.length) {
      for (j <- i + 1 until strs.length) {
        var similar = similarity(strs(i), strs(j))
        if (map.contains(strs(i)))
          map += (strs(i) -> (map.get(strs(i)).get + similar))
        else
          map += (strs(i) -> similar)
        if (map.contains(strs(j)))
          map += (strs(j) -> (map.get(strs(j)).get + similar))
        else
          map += (strs(j) -> similar)
      }
    } //end of for
    if (map.size > 0)
      map.toSeq.sortWith(_._2 > _._2)(0)._1
    else
      ""
  }
}
